/* 
 * Copyright (C), 2016-2020, 厦门盈欣宝金融信息服务有限公司
 * Project Name :boot-parent
 * File Name: org.easytouseit.rabbitmq.config
 * Encoding UTF-8 
 * Version: 1.0 
 * Date: 2018-09-14 16:19
 * History:  
 * 1. Date: 2018-09-14
 *    Author: guojm 
 *    Modification: 新建
 * 2. ...
 */
package org.easytouseit.rabbitmq.config;

import org.easytouseit.rabbitmq.callback.MsgSendReturnCallback;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

import javax.annotation.Resource;

/**
 * @author guojm
 * @version V1.0
 * @date 2018-09-14 下午 16:19
 */
@Configuration
public class CommonRabbitMQConfig {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    public final static String QUEUE_NAME = "direct-queue";
    public final static String EXCHANGE_NAME = "direct-exchange";
    public final static String ROUTING_KEY = "spring-boot-key";
    @Autowired
    private RabbitProperties rabbitProperties;
    /*// 创建队列
    @Bean("queueDirect")
    public Queue queueDirect() {
        return QueueBuilder.durable(QUEUE_NAME).build();
    }
    // 创建一个 topic 类型的交换器
    @Bean("exchangeDirect")
    public DirectExchange exchangeDirect() {
        return new DirectExchange(EXCHANGE_NAME,true,false);
    }

    // 使用路由键（routingKey）把队列（Queue）绑定到交换器（Exchange）
    @Bean
    public Binding binding(@Qualifier("queueDirect") Queue queueDirect,
                           @Qualifier("exchangeDirect") DirectExchange exchangeDirect) {
        return BindingBuilder.bind(queueDirect).to(exchangeDirect).with(ROUTING_KEY);
    }
    public final static String TOPIC_QUEUE_ONE = "topic.queue.one";
    public final static String TOPIC_QUEUE_TWO = "topic.queue.two";
    public final static String TOPIC_EXCHANGE_NAME = "topic-exchange";
    @Bean("topicQueueOne")
    public Queue topicQueueOne() {
        return QueueBuilder.durable(TOPIC_QUEUE_ONE).build();
    }
    @Bean("topicQueueTwo")
    public Queue topicQueueTwo() {
        return QueueBuilder.durable(TOPIC_QUEUE_TWO).build();
    }

    @Bean("exchangeTopic")
    public TopicExchange exchangeTopic() {
        return new TopicExchange(TOPIC_EXCHANGE_NAME,true,false);
    }

    @Bean
    public Binding bindingTopicQueueOne(@Qualifier("topicQueueOne") Queue topicQueueOne,
                                        @Qualifier("exchangeTopic") TopicExchange exchangeTopic) {
        return BindingBuilder.bind(topicQueueOne).to(exchangeTopic).with("topic.queue.one");
    }
    @Bean
    public Binding bindingTopicQueueTwo(@Qualifier("topicQueueTwo") Queue topicQueueTwo,
                                        @Qualifier("exchangeTopic") TopicExchange exchangeTopic) {
        return BindingBuilder.bind(topicQueueTwo).to(exchangeTopic).with("topic.queue.#");
    }*/
    public final static String FANOUT_QUEUE_ONE = "fanout.queue.one";
    public final static String FANOUT_QUEUE_TWO = "fanout.queue.two";
    public final static String FANOUT_QUEUE_THREE = "fanout.queue.three";
    public final static String FANOUT_EXCHANGE_NAME = "fanout-exchange";
    public final static String FANOUT_ROUTING_KEY = "fanout.key";

    /*@Bean("fanoutQueueOne")
    public Queue fanoutQueueOne() {
        return QueueBuilder.durable(FANOUT_QUEUE_ONE).build();
    }
    @Bean("fanoutQueueTwo")
    public Queue fanoutQueueTwo() {
        return QueueBuilder.durable(FANOUT_QUEUE_TWO).build();
    }
    @Bean("fanoutQueueThree")
    public Queue fanoutQueueThree() {
        return QueueBuilder.durable(FANOUT_QUEUE_THREE).build();
    }

    @Bean("exchangeFanout")
    public FanoutExchange  exchangeFanout() {
        return new FanoutExchange(FANOUT_EXCHANGE_NAME,true,false);
    }

    @Bean
    public Binding bindingFanoutQueueOne(@Qualifier("fanoutQueueOne") Queue fanoutQueueOne,
                                         @Qualifier("exchangeFanout") FanoutExchange exchangeFanout) {
        return BindingBuilder.bind(fanoutQueueOne).to(exchangeFanout);
    }
    @Bean
    public Binding bindingFanoutQueueTwo(@Qualifier("fanoutQueueTwo") Queue fanoutQueueTwo,
                                         @Qualifier("exchangeFanout") FanoutExchange exchangeFanout) {
        return BindingBuilder.bind(fanoutQueueTwo).to(exchangeFanout);
    }
    @Bean
    public Binding bindingFanoutQueueThree(@Qualifier("fanoutQueueThree") Queue fanoutQueueThree,
                                           @Qualifier("exchangeFanout") FanoutExchange exchangeFanout) {
        return BindingBuilder.bind(fanoutQueueThree).to(exchangeFanout);
    }*/
    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(rabbitProperties.getHost(), rabbitProperties.getPort());
        connectionFactory.setUsername(rabbitProperties.getUsername());
        connectionFactory.setPassword(rabbitProperties.getPassword());
        connectionFactory.setVirtualHost(rabbitProperties.getVirtualHost());
        connectionFactory.setPublisherConfirms(rabbitProperties.isPublisherConfirms());
        connectionFactory.setPublisherReturns(rabbitProperties.isPublisherReturns());
        return connectionFactory;
    }
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        return factory;
    }
    /**
     * 重试机制
     * @return
     */
    @Bean
    public RetryTemplate retryTemplate(){
        RetryTemplate retryTemplate = new RetryTemplate();
        ExponentialBackOffPolicy backOffPolicy  = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(500); //initialInterval是重试的重试间隔时间
        backOffPolicy.setMultiplier(10.0); //multiplier是从初始值到最大的增长倍数，其实我们还可以配置SimpleRetryPolicy来指定重试的次数
        backOffPolicy.setMaxInterval(10000); //maxInterval是重试最大间隔时间
        retryTemplate.setBackOffPolicy(backOffPolicy);
        //默认的retryPolicy当抛出Exception异常的时候，重试3次，可自定义，详情查看RetryTemplate中的配置。
        //retryTemplate.setRetryPolicy(new SimpleRetryPolicy());
        return retryTemplate;
    }
    /**
     * 定制化amqp模版      可根据需要定制多个
     * <p>
     * <p>
     * 此处为模版类定义 Jackson消息转换器
     * ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调   即消息发送到exchange  ack
     * ReturnCallback接口用于实现消息发送到RabbitMQ 交换器，但无相应队列与交换器绑定时的回调  即消息发送不到任何一个队列中  ack
     *
     * @return the amqp template
     */
    // @Primary
    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate  rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate =new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        rabbitTemplate.setEncoding("UTF-8");
        // 消息发送失败返回到队列中，yml需要配置 publisher-returns: true
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(new MsgSendReturnCallback());

        // 消息确认，yml需要配置 publisher-confirms: true
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                logger.info("消息发送到exchange成功,id: {}", correlationData.getId());
            } else {
                logger.info("消息发送到exchange失败,原因: {}", cause);
            }
        });
        //添加重试
        rabbitTemplate.setRetryTemplate(retryTemplate());
        return rabbitTemplate;
    }


}
